package org.lhl.readruler.connection

import com.blankj.utilcode.util.LogUtils
import io.reactivex.Observable
import io.reactivex.disposables.Disposable
import io.reactivex.processors.FlowableProcessor
import lhl.kotlinextends.toastLong
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.TimeUnit

abstract class AbstractConnect(private val publish: FlowableProcessor<Any>, private val parser: Parser?) {

    private val TIME_OUT = 20000L

    var connectState = ConnectState.DISCONNECTED
    private val mPacketsToSend = LinkedBlockingQueue<ByteArray>()
    private var mConnectingThread: Thread? = null

    private var connectTimeout: Disposable? = null

    protected abstract fun openConnection()
    protected abstract fun closeConnection()
    protected abstract fun sendBuffer(byteArray: ByteArray)
    protected abstract fun readDataBlock(byteArray: ByteArray): Int

    fun connect() {
        if (connectState == ConnectState.CONNECTING) {
            return
        }
        connectTimeout?.dispose()
        connectTimeout = Observable.timer(TIME_OUT, TimeUnit.MILLISECONDS).subscribe {
            if (connectState == ConnectState.CONNECTING) {
                disConnect()
                publish.onNext(AttEvent(Event.CONNECT_TIMEOUT))
            }
        }
        mConnectingThread?.run { if (isAlive) interrupt() }
        connectState = ConnectState.CONNECTING
        publish.onNext(AttEvent(Event.CONNECTING))
        mConnectingThread = Thread(connectingRunnable, "connecting-thread").apply { start() }
    }

    fun disConnect() {
        LogUtils.e("蓝牙测试  disConnect")
        if (connectState == ConnectState.DISCONNECTED) return
        connectState = ConnectState.DISCONNECTED
        publish.onNext(AttEvent(Event.DISCONNECTED))
        mConnectingThread?.run { if (isAlive) interrupt() }
        closeConnection()
    }

    /**
     * 连接与接收线程
     */
    private val connectingRunnable = Runnable {
        var sendThread: Thread? = null
        try {
            openConnection()
            sendThread = Thread(sendRunnable, "send-thread").apply { start() }
            val readBuffer = ByteArray(4096)
            while (connectState == ConnectState.CONNECTED || connectState == ConnectState.CONNECTING) {
                val bufferSize = readDataBlock(readBuffer)
                handleData(bufferSize, readBuffer)
            }
        } catch (e: Exception) {
            LogUtils.e("蓝牙测试  连接线程 = $e")
            sendThread?.run { if (isAlive) interrupt() }
            disConnect()
        }
    }

    @Synchronized
    fun send(byteArray: ByteArray) {
        if (byteArray == null) return
        if (connectState != ConnectState.CONNECTED) return
        if (!mPacketsToSend.offer(byteArray)) {
            "发送缓冲区满".toastLong()
        }
    }

    /**
     * 发送线程
     */
    private val sendRunnable = Runnable {
        try {
            while (connectState == ConnectState.CONNECTING) Thread.sleep(100)
            while (connectState == ConnectState.CONNECTED) {
                val buffer = mPacketsToSend.take()
                sendBuffer(buffer)
            }
        } catch (e: Exception) {
            LogUtils.e("蓝牙测试 发送线程 = $e")
            disConnect()
        }
    }


    private fun handleData(bufferSize: Int, buffer: ByteArray) {
        if (bufferSize < 1) return
        for (i in 0 until bufferSize) {
            parser?.parse(buffer[i].toInt() and 0xff)?.also(publish::onNext)
        }
    }


}